package org.snail.proxy.client;

import com.google.protobuf.ByteString;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.snail.proxy.proto.*;

public class LocalTCPClient implements Client {
    private static final Logger log = LoggerFactory.getLogger(LocalTCPClient.class);

    private EventLoopGroup group;
    private Channel channel;

    private final Channel proxyChannel;
    private final String id;

    public LocalTCPClient(Channel proxyChannel, String id) {
        this.proxyChannel = proxyChannel;
        this.id = id;
    }

    @Override
    public void connect(String ip, int port) {
        try {
            if (group != null) {
                group.shutdownGracefully();
                group = null;
            }
            if (channel != null) {
                channel.closeFuture();
                channel = null;
            }

            group = new NioEventLoopGroup();
            // 发起异步连接请求，绑定连接端口和host信息，使用NioSocketChannel来作为连接用的channel类
            ChannelFuture future = new Bootstrap().group(group).channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() { // 绑定连接初始化器
                        @Override
                        public void initChannel(SocketChannel ch) {
                            ch.pipeline()
                                    .addLast(new BytesDecoder())
                                    .addLast(new BytesEncoder())
                                    .addLast(new ChannelInboundHandlerAdapter() {
                                        @Override
                                        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                                            DeleteClientRequest request = DeleteClientRequest.newBuilder()
                                                    .setId(id)
                                                    .build();
                                            proxyChannel.writeAndFlush(new Packet((byte) 0, MessageId.DELETE_CLIENT_REQUEST_VALUE, request.toByteArray()));
                                            super.channelInactive(ctx);
                                            log.info("Local server channelInactive: {}.", id);
                                        }

                                        @Override
                                        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                            byte[] data = (byte[]) msg;
                                            ProxyMessage message = ProxyMessage.newBuilder()
                                                    .setId(id)
                                                    .setData(ByteString.copyFrom(data))
                                                    .build();
                                            proxyChannel.writeAndFlush(new Packet((byte) 0, MessageId.PROXY_MESSAGE_VALUE, message.toByteArray()));
                                            super.channelRead(ctx, msg);
                                            log.info("Local tcp server message: {}[{}].", id, data.length);
                                        }
                                    });
                        }
                    }).connect(ip, port)
                    .sync();

            future.addListener((ChannelFutureListener) arg0 -> {
                if (future.isSuccess()) {
                    log.info("Connect local server success: {}:{}.", ip, port);
                } else {
                    log.error("Connect proxy server failed: {}:{}.", ip, port);
                    future.cause().printStackTrace();
                    group.shutdownGracefully(); //关闭线程组
                }
            });

            this.channel = future.channel();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void disconnect() {
        channel.close();
    }

    @Override
    public void sendMessage(Object message) {
        if (channel == null || message == null || !channel.isWritable() || !channel.isActive()) {
            return;
        }
        channel.writeAndFlush(message);
        log.info("LocalTCPClient sendMessage: {}.", message.getClass().getSimpleName());
    }

    @Override
    public boolean isConnected() {
        return channel != null && channel.isActive();
    }
}
